
        #ifndef __M_SERVER_H__
#define __M_SERVER_H__
#include<iostream>
#include<vector>
#include<cassert>
#include<string>
#include<cstdint>
#include<string.h>
#include<ctime>
#include<sys/socket.h>
#include<arpa/inet.h>
#include<fcntl.h>
#include<unistd.h>
#include<netinet/in.h>
#include<functional>
#include<sys/epoll.h>
#include<unordered_map>
#include<mutex>
#include<condition_variable>
#include<thread>
#include<memory>
#include<sys/eventfd.h>
#include<sys/timerfd.h>
#include<typeinfo>
#include<signal.h>
#include<fstream>

#define BUFFER_DEFAULT_SIZE 1024

#define INF 0
#define DBG 1
#define ERR 2

#define LOG_LEVEL ERR

#define LOG(level,format,...) do{\
    if(level<LOG_LEVEL)break;\
    time_t t = time(NULL);\
    struct tm *ltm = localtime(&t);\
    char tmp[32] = {0};\
    strftime(tmp,31,"%H:%M:%S",ltm);\
    fprintf(stdout,"[%p %s %s:%d] " format "\n", (void*)pthread_self(),tmp,__FILE__,__LINE__,##__VA_ARGS__);\
} while (0)

#define INF_LOG(format,...) LOG(INF,format,##__VA_ARGS__)
#define DBG_LOG(format,...) LOG(DBG,format,##__VA_ARGS__)
#define ERR_LOG(format,...) LOG(ERR,format,##__VA_ARGS__)

class Buffer
{
    private:
        std::vector<char> _buffer;  //使用vector进行内存空间管理
        uint64_t _reader_idx;       //读偏移
        uint64_t _writer_idx;       //读偏移

    public:
        Buffer():_reader_idx(0),_writer_idx(0),_buffer(BUFFER_DEFAULT_SIZE){}

        char* Begin()
        {
            return &*_buffer.begin();
        }

        //获取当前写入起始地址 buffer的空间起始地址，加上写偏移量
        char *WritePosition()
        {
            return Begin()+_writer_idx;
        }

        //获取当前读取起始地址
        char *ReadPosition()
        {
            return Begin()+_reader_idx;
        }

        //获取前沿(缓冲区末尾)空闲空间大小--写偏移之后的空闲空间
        uint64_t TailIdleSize()
        {
            return _buffer.size()-_writer_idx;
        }

        //获取后沿(缓冲区起始)空闲空间大小--读偏移之前的空闲空间
        uint64_t HeadIdleSize()
        {
            return _reader_idx;
        }

        //获取可读数据大小
        uint64_t ReadAbleSize()
        {
            return _writer_idx-_reader_idx;
        }

        //将读偏移向后移动
        void MoveReadOffset(uint64_t len)
        {
            if(len == 0)
                return;
            //向后移动大小必须小于当前可读数据空间大小
            assert( len<=ReadAbleSize());
            _reader_idx+=len;
        }

        //将写偏移向后移动
        void MoveWriteOffset(uint64_t len)
        {
            //向后移动大小必须小于当前后边的空闲空间大小
            assert(len<=TailIdleSize());
            _writer_idx+=len;
        }

        //确保可写空间足够（整体空闲空间够了就移动数据，否则就扩容）
        void EnsureWriteSpace(uint64_t len)
        {
            //如果末尾空闲空间大小足够，则直接返回
            if(TailIdleSize()>=len)
                return;
            //末尾空闲空间不够则加上前置位置，将数据移动到起始位置
            if(len<=TailIdleSize()+HeadIdleSize())
            {
                uint64_t rsz = ReadAbleSize();
                std::copy(ReadPosition(), ReadPosition() + rsz, Begin());//把可读数据拷贝到起始位置
                _reader_idx = 0;        //将读偏移归零
                _writer_idx = rsz;      //将写位置置位可读数据大小，因为起始为0
            }
            else
            {
                //总体空间不够，则需要扩容，不移动数据，直接给写偏移之后扩容足够空间即可
                _buffer.resize(_writer_idx+len);
            }
        }

        //写入数据
        void Write(const void* data,uint64_t len)
        {
            //保证有足够空间，拷贝数据
            if(len==0)
                return;
            EnsureWriteSpace(len);
            const char* d = (const char*) data;
            std::copy(d, d+len, WritePosition());
        }

        void WriteAndPush(const void* data,uint64_t len)
        {
            Write(data,len);
            MoveWriteOffset(len);
        }

        void WriteString(const std::string &data)
        {
            Write(data.c_str(),data.size());
        }

        void WriteStringAndPush(const std::string &data)
        {
            WriteString(data);
            MoveWriteOffset(data.size());
        }

        void WriteBuffer(Buffer &data)
        {
            Write(data.ReadPosition(),data.ReadAbleSize());
        }

        void WriteBufferAndPush(Buffer &data)
        {
            WriteBuffer(data);
            MoveWriteOffset(data.ReadAbleSize());
        }

        //读取数据
        void Read(void* buf,uint64_t len)
        {
            //要获取的数据大小要小于可读数据大小
            assert(len<=ReadAbleSize());

            std::copy(ReadPosition(),ReadPosition()+len,(char*)buf);
        }

        void ReadAndPop(void* buf,uint64_t len)
        {
            Read(buf,len);
            MoveReadOffset(len);
        }

        std::string ReadAsString(uint64_t len)
        {
            assert(len<=ReadAbleSize());
            std::string str;
            str.resize(len);
            Read(&str[0],len);
            return str;
        }

        std::string ReadAsStringAndPop(uint64_t len)
        {
            std::string str = ReadAsString(len);
            MoveReadOffset(len);
            return str;
        }

        char* FindCRLF()
        {
            char* res =(char*)memchr(ReadPosition(),'\n',ReadAbleSize());
            return res;
        }
        
        //通常获取一行数据，这种情况针对是
        std::string GetLine()
        {
            char* pos = FindCRLF();
            if(pos==NULL)
                return "";

            //+1是为了取出换行
            return ReadAsString(pos-ReadPosition()+1);
        }

        std::string GetLineAndPop()
        {
            std::string str = GetLine();
            MoveReadOffset(str.size());
            return str;
        }

        //清空缓冲区
        void Clear()
        {
            _writer_idx = 0;
            _reader_idx = 0;
        }
};

#define MAX_LISTEN 1024

class Socket
{
    private:
        int _sockfd;
    public:
        Socket():_sockfd(-1){}
        Socket(int fd):_sockfd(fd){}
        ~Socket(){Close();}

        int Fd(){return _sockfd;}
        //创建套接字
        bool Create()
        {
            _sockfd = socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
            if(_sockfd<0)
            {
                ERR_LOG("CREATE SOCKET FAILED!!");
                    return false;
            }
            return true;
        }

        //绑定地址信息
        bool Bind(const std::string &ip,uint16_t port)
        {
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            socklen_t len = sizeof(struct sockaddr_in);
            int ret = bind(_sockfd,(struct sockaddr*)&addr,len);
            if(ret<0)
            {
                ERR_LOG("BIND ADDRESS FAILED!");
                    return false;
            }
            return true;
        }

        //开始监听
        bool Listen(int backlog = MAX_LISTEN)
        {
            int ret = listen(_sockfd,backlog);
            if(ret<0)
            {
                ERR_LOG("SOCKET LISTEN FAILED!");
                    return false;
            }
            return true;
        }

        //向服务器发起连接
        bool Connect(const std::string &ip,uint16_t port)
        {
            struct sockaddr_in addr;
            addr.sin_family = AF_INET;
            addr.sin_port = htons(port);
            addr.sin_addr.s_addr = inet_addr(ip.c_str());
            socklen_t len = sizeof(struct sockaddr_in);
            int ret = connect(_sockfd,(struct sockaddr*)&addr,len);
            if(ret<0)
            {
                ERR_LOG("CONNECT SERVER FAILED!");
                    return false;
            }
            return true;
        }

        //获取新连接
        int Accept()
        {
            int newfd = accept(_sockfd,NULL,NULL);
            if(newfd<0)
            {
                ERR_LOG("SOCKET ACCEPT FAILED!");
                return -1;
            }
            return newfd;
        }

        //接收数据
        ssize_t Recv(void* buf,size_t len,int flag =0)
        {
            ssize_t ret = recv(_sockfd,buf,len,flag);
            if(ret<=0)
            {
                //EAGAIN 表示没有数据了，非阻塞情况下才出现
                //EINTR  表示当前socket的阻塞等待被信号打断了
                if(errno ==EAGAIN||errno==EINTR)
                    return 0;
                ERR_LOG("SOCKET RECV FAILED!");
                return -1;
            }
            return ret;
        }

        //非阻塞接收
        ssize_t NonBlockRecv(void* buf,size_t len)
        {
            return Recv(buf,len,MSG_DONTWAIT);
        }

        //发送数据
        ssize_t Send(const void* buf,size_t len,int flag = 0)
        {
            ssize_t ret = send(_sockfd,buf,len,flag);
            if(ret<0)
            {
                if(errno ==EAGAIN||errno==EINTR)
                    return 0;
                ERR_LOG("SOCKET SEND FAILED!");
                return -1;
            }
            return ret;
        }

        //非阻塞发送
        ssize_t NonBlockSend(void* buf,size_t len)
        {
            if(len==0)return 0;
            return Send(buf,len,MSG_DONTWAIT);
        }

        //关闭套接字
        void Close()
        {
            if(_sockfd !=-1)
            {
                close(_sockfd);
                _sockfd = -1;
            }
        }

        //创建一个服务端连接
        bool CreateServer(uint16_t port,const std::string &ip = "0.0.0.0",bool block_flag = 0)
        {
            if(Create() == false)
                return false;
            if(Bind(ip,port)==false)
                return false;
            if(Listen()==false)
                return false;
            if(block_flag)NonBlock();
            ReuseAddress();

            return true;
        }

        //创建提供客户端连接
        bool CreateClient(uint16_t port,const std::string &ip)
        {
            if(Create() == false)
                return false;
            if(Connect(ip,port)==false)
                return false;

            return true;
        }

        //设置套接字选项--开启地址端口重用
        void ReuseAddress()
        {
            int val = 1;
            setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,(void*)&val,sizeof(int));
            val = 1;
            setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,(void*)&val,sizeof(int));
        }

        //设置套接字阻塞属性--设置为非阻塞
        void NonBlock()
        {
            int flag = fcntl(_sockfd,F_GETFL,0);
            fcntl(_sockfd,F_SETFL,flag|O_NONBLOCK);
        }
};


class Poller;
class EventLoop;
class Channel
{
    private:
        int _fd;
        EventLoop *_loop;
        uint32_t _events;       //当前需要监控的事件
        uint32_t _revents;      //当前连接触发的事件
        using EventCallback = std::function<void()>;
        EventCallback _read_callback;       //可读事件触发回调函数
        EventCallback _write_callback;      //可写事件触发回调函数
        EventCallback _error_callback;      //错误事件触发回调函数
        EventCallback _close_callback;      //连接关闭事件触发回调函数
        EventCallback _event_callback;      //任意事件触发回调函数

    public:
        Channel(EventLoop* loop,int fd):
            _fd(fd),_events(0),_revents(0),_loop(loop){}

        int Fd()
        {
            return _fd;
        }

        void SetRevents(uint32_t events)
        {
            _revents = events;
        }

        uint32_t Events()
        {
            return _events;
        }

        void SetReadCallback(const EventCallback& cb)
        {
            _read_callback = cb;
        }

        void SetWriteCallback(const EventCallback& cb)
        {
            _write_callback = cb;
        }

        void SetErrorCallback(const EventCallback& cb)
        {
            _error_callback = cb;
        }

        void SetCloseCallback(const EventCallback& cb)
        {
            _close_callback = cb;
        }

        void SetEventCallback(const EventCallback& cb)
        {
            _event_callback = cb;
        }
        //当前是否监控了可读
        bool ReadAble()         
        {
            return (_events & EPOLLIN);
        }

        //当前是否监控了可写
        bool WriteAble()        
        {
            return (_events & EPOLLOUT);
        }

        //启动读事件监控
        void EnableRead()       
        {
            _events |= EPOLLIN;
            Update();
        }

        //启动写事件监控
        void EnableWrite()
        {
            _events |= EPOLLOUT;
            Update();
        }  

        //关闭读事件监控
        void DisableRead()
        {
            _events &= ~EPOLLIN;
            Update();
        }     

        //关闭写事件监控
        void DisableWrite()
        {
            _events &= ~EPOLLOUT;
            Update();
        } 

        //关闭所有事件监控
        void DisableAll()
        {
            _events = 0;
        }

        //移除监控
        void Remove();

        void Update();
        
        //事件处理，一旦触发了事件就调用,自己触发了什么事件如何处理自己决定
        void HandleEvent()
        {
            if((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents& EPOLLPRI))
            {
                //不管任何事件都调用
                if(_read_callback)
                    _read_callback();
            }

            //有可能释放连接的操作，一次只处理一个
            if(_revents & EPOLLOUT)
            {
                if(_write_callback)
                    _write_callback();
            }
            else if(_revents & EPOLLERR)
            {
                if(_error_callback)
                    _error_callback();
            }
            else if(_revents & EPOLLHUP)
            {
                if(_close_callback)
                    _close_callback();
            }
            if(_event_callback)//刷新活跃度
                _event_callback();
        }
};


#define MAX_EPOLLEVENTS 1024

class Poller
{
    private:
        int _epfd;
        struct epoll_event _evs[MAX_EPOLLEVENTS];
        std::unordered_map<int,Channel*> _channels;

    private:

        //对epoll的直接操作
        void Update(Channel *channel,int op)
        {
            int fd = channel->Fd();
            struct epoll_event ev;
            ev.data.fd = fd;
            ev.events = channel->Events();
            int ret = epoll_ctl(_epfd,op,fd,&ev);
            if(ret<0)
            {
                ERR_LOG("EPOLLCTL FAILED");
                abort();//退出程序
            }

            return;
        }

        //判断一个channel是否已经添加了事件监控
        bool HasChannel(Channel *channel)
        {
            auto it = _channels.find(channel->Fd());
            if(it == _channels.end())
                return false;
            return true;
        }
    
    public:
        Poller()
        {
            _epfd = epoll_create(MAX_EPOLLEVENTS);

            if(_epfd<0)
            {
                ERR_LOG("EPOLL CREATE FAILED");
                abort();
            }
        }

        //添加或修改监控事件
        void UpdateEvent(Channel* channel)
        {
            bool ret = HasChannel(channel);
            if(ret==false)
            {
                _channels.insert(std::make_pair(channel->Fd(),channel));
                return Update(channel,EPOLL_CTL_ADD);
            }
            return Update(channel,EPOLL_CTL_MOD);
        }
        
        //移除监控
        void RemoveEvent(Channel* channel)
        {
            auto it = _channels.find(channel->Fd());
            if(it!=_channels.end())
            {
                _channels.erase(it);
            }

            Update(channel,EPOLL_CTL_DEL);
        }

        //开始监控，返回活跃连接
        void Poll(std::vector<Channel*>* active)
        {
            int nfds = epoll_wait(_epfd,_evs,MAX_EPOLLEVENTS,-1);
            if(nfds<0)
            {
                if(errno ==EINTR)
                {
                    return;
                }
                ERR_LOG("EPOLL WAIT ERROR:%s\n",strerror(errno));
                abort();
            }

            for(int i = 0;i<nfds;i++)
            {
                auto it = _channels.find(_evs[i].data.fd);
                assert(it!=_channels.end());
                it->second->SetRevents(_evs[i].events);
                active->push_back(it->second);
            }

            return;
        }
};


using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimerTask
{
    private:
        uint64_t _id;           //定时器任务对象id
        uint32_t _timeout;      //定时任务的超时时间
        bool _canceled;         //false - 没有被取消  true - 被取消了不用执行
        TaskFunc _task_cb;      //定时器对象要执行的定时任务
        ReleaseFunc _release;   //用于删除timerwheel中保存到定时器对象信息
    public:

        TimerTask(uint64_t id,uint32_t delay,const TaskFunc &cb):
        _id(id),_timeout(delay),_task_cb(cb),_canceled(false){}

        ~TimerTask()
        {
            if(_canceled==false)
                _task_cb();
            _release();
        }

        void Cancel()
        {
            _canceled = true;
        }

        void SetRelease(const ReleaseFunc &cb)
        {
            _release = cb;
        }

        uint32_t DelayTime()
        {
            return _timeout;
        }
};



class TimerWheel
{
    private:
        int _capacity;      //表盘最大数量  最大延迟时间
        using WeakTask = std::weak_ptr<TimerTask>;
        using PtrTask = std::shared_ptr<TimerTask>;
        std::vector<std::vector<PtrTask>> _wheel;
        std::unordered_map<uint64_t,WeakTask> _timers;

        int _tick;          //当前的秒针，走到哪里释放哪里，即执行哪里的任务

        EventLoop * _loop;
        int _timerfd;       //定时器描述符 可读事件就是读取计数器，执行定时任务
        std::unique_ptr<Channel> _timer_channel;

    private:
        void RemoveTimer(uint64_t id)
        {
            auto it = _timers.find(id);
            if(it!=_timers.end())
                _timers.erase(it);
        }

        static int CreateTimerfd()
        {
            int timerfd = timerfd_create(CLOCK_MONOTONIC,0);
            if(timerfd<0)
            {
                ERR_LOG("TIMERFD CREATE FAILED");
                abort();
            }

            struct itimerspec itime;
            itime.it_value.tv_sec = 1;
            itime.it_value.tv_nsec = 0;
            itime.it_interval.tv_sec = 1;
            itime.it_interval.tv_nsec = 0;
            timerfd_settime(timerfd,0,&itime,NULL);
            return timerfd;
        }

        int ReadTimerfd()
        {
            uint64_t times;
            //有可能因为其他描述符的事件处理花费时间较长 再处理定时器任务时可能超时了很多次
            //read读取到的数据times就是从上一次read之后超时的次数
            int ret = read(_timerfd,&times,8);
            if(ret<0)
            {
                ERR_LOG("READ TIMERFD FAILED");
                abort();
            }

            return times;
        }

        void RunTimerTask()//每秒执行一次
        {
            _tick = (_tick+1)%_capacity; 
            _wheel[_tick].clear();  //清空指定位置的数组，释放数组内的定时器对象的shared_ptr，释放完为0则执行对应的回调
        }

        void OnTime()
        {
            //根据实际超时的次数执行对应的超市任务
            int times = ReadTimerfd();
            for(int i = 0;i<times;i++)
                RunTimerTask();
        }

        void TimerAddInLoop(uint64_t id,uint32_t delay,const TaskFunc &cb)//添加定时任务
        {
            PtrTask pt(new TimerTask(id,delay,cb));
            pt->SetRelease(std::bind(&TimerWheel::RemoveTimer,this,id));

            int pos = (_tick+delay)%_capacity;
            _wheel[pos].push_back(pt);            
            _timers[id] = WeakTask(pt);
        }

        void TimerRefreshInLoop(uint64_t id)//刷新延时定时任务
        {
            auto it = _timers.find(id);
            if(it==_timers.end())
                return;
            PtrTask pt = it->second.lock();
            if(pt == nullptr)
                return;

            int delay = pt->DelayTime();
            int pos = (_tick+delay)%_capacity;
            _wheel[pos].push_back(pt);
        }

        void TimerCancelInLoop(u_int64_t id)
        {
            auto it = _timers.find(id);
            if(it==_timers.end())
                return;
            PtrTask pt = it->second.lock();
            if(pt)pt->Cancel();
        }

    public:
        TimerWheel(EventLoop* loop):_capacity(60),_tick(0),_wheel(_capacity),_loop(loop),_timerfd(CreateTimerfd()),_timer_channel(new Channel(_loop,_timerfd))
        {
            _timer_channel->SetReadCallback(std::bind(&TimerWheel::OnTime,this));
            _timer_channel->EnableRead();//启动读事件监控
        }

        void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc &cb);//考虑线程安全 _timers的操作有可能在多线程中进行，不想加锁就放在同一个线程中

        void TimerRefresh(uint64_t id);

        void TimerCancel(u_int64_t id);

        bool HasTimer(uint64_t id)//存在线程安全问题 只能在模块内调用
        {
            auto it = _timers.find(id);
            if(it==_timers.end())
                return false;
            return true;
        }
};


class EventLoop
{
    private:
        using Functor = std::function<void()>;
        std::thread::id _thread_id;     //线程id
        int _event_fd;                  //eventfd唤醒IO事件监控有可能导致的阻塞
        std::unique_ptr<Channel> _event_channel;
        Poller _poller;                 //进行所有描述符的事件监控
        std::vector<Functor> _tasks;    //任务池
        std::mutex _mutex;              //实现任务池操作的线程安全
        TimerWheel _timer_wheel;

    public:
        void RunAllTask()                       //执行任务池中的所有任务
        {
            std::vector<Functor> functor;
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _tasks.swap(functor);
            }

            for(auto &f: functor)
            {
                f();
            }

            return;
        }

        static int CreateEventFd()
        {
            int efd = eventfd(0,EFD_CLOEXEC|EFD_NONBLOCK);
            if(efd<0)
            {
                ERR_LOG("EVENTFD FAILED");
                abort();
            }

            return efd;
        }

        void ReadEventfd()
        {
            uint64_t res = 0;
            int ret = read(_event_fd,&res,sizeof(res));
            if(ret<0)
            {
                //EINTR 被信号打断  EAGAIN 无数据可读
                if(errno==EINTR||errno==EAGAIN)
                {
                    return;
                }
                ERR_LOG("READ EVENTFD FAILED");
                abort();
            }
            return;
        }

        void WeakUpEventFd()
        {
            uint64_t val = 1;
            int ret = write(_event_fd,&val,sizeof(val));
            if(ret<0)
            {
                if(errno==EINTR)
                {
                    return;
                }
                ERR_LOG("READ EVENTFD FAILED");
                abort();
            }
            return;
        }

    public:
        EventLoop():_thread_id(std::this_thread::get_id()),
                    _event_fd(CreateEventFd()),
                    _event_channel(new Channel(this,_event_fd)),
                    _timer_wheel(this)
                    {   
                        //给eventfd添加可读事件回调函数 读取eventfd事件通知次数
                        _event_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd,this));
                        //启动eventfd的读事件监控
                        _event_channel->EnableRead();
                    }

        void Start()                          
        {
            //三步走 事件监控 就绪事件处理 执行任务
            while(1)
            {
                std::vector<Channel*> actives;
                _poller.Poll(&actives);

                for(auto& channel:actives)
                {
                    channel->HandleEvent();
                }

                RunAllTask();
            }
        }

        
        bool IsInLoop()                         //用于判断当前线程是否是eventloop对应线程
        {
            return _thread_id==std::this_thread::get_id();
        }

        void AssertInLoop()
        {
            assert(_thread_id==std::this_thread::get_id());
        }

        void RunInLoop(const Functor &cb)       //判断将要执行的任务是否处于当前线程中，是则执行，否则加入任务池
        {
            if(IsInLoop())
            {
                return cb();
            }
            return QueueInLoop(cb);
        }

        void QueueInLoop(const Functor &cb)     //将操作压入任务池
        {
            {
                std::unique_lock<std::mutex> _lock(_mutex);
                _tasks.push_back(cb);
            }
            //唤醒可能因为没有事件就绪，而导致的epoll阻塞
            //给eventfd写入一个数据，唤醒eventfd就触发可读事件
            WeakUpEventFd();
        }

        void UpdateEvent(Channel* channel)      //添加或修改描述符的事件监控
        {
            return _poller.UpdateEvent(channel);
        }

        void RemoveEvent(Channel* channel)      //移除描述符的监控
        {
            return _poller.RemoveEvent(channel);
        }

        void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc &cb)
        {
            return _timer_wheel.TimerAdd(id,delay,cb);
        }

        void TimerRefresh(uint64_t id)
        {
            return _timer_wheel.TimerRefresh(id);
        }

        void TimerCancel(uint64_t id)
        {
            return _timer_wheel.TimerCancel(id);
        }

        bool HasTimer(uint64_t id)
        {
            return _timer_wheel.HasTimer(id);
        }
};

class LoopThread
{
    private:
        //用于实现_loop获取的同步关系 防止线程创建了 但loop还未实例化之前获取loop
        std::mutex _mutex;          //互斥锁
        std::condition_variable _cond;   //条件变量
        EventLoop* _loop;           //Eventloop指针变量，需要在线程内实例化
        std::thread _thread;        //Eventloop对应的线程

    private:
        //实例化EventLoop对象 唤醒cond上可能有阻塞的线程 并且开始运行Eventloop模块的功能
        void ThreadEntry()
        {
            EventLoop loop;
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _loop = &loop;
                _cond.notify_all();
            }
            loop.Start();
        }

    public:
        //创建线程 设定线程入口函数
        LoopThread():_loop(NULL),_thread(std::thread(&LoopThread::ThreadEntry,this)){}

        //返回当前线程关联的Eventloop对象指针
        EventLoop* GetLoop()
        {
            EventLoop *loop = NULL;
            {
                std::unique_lock<std::mutex> lock(_mutex);      //加锁
                _cond.wait(lock,[&](){return _loop!=NULL;});    //loop为空则循环阻塞
                loop =_loop; 
            }

            return loop;
        }

};


class LoopThreadPool
{
    private:
        int _thread_count;
        int _next_idx;
        EventLoop*_baseloop;
        std::vector<LoopThread*> _threads;
        std::vector<EventLoop*> _loops;

    public:
        LoopThreadPool(EventLoop* baseloop):_thread_count(0),_next_idx(0),_baseloop(baseloop){}

        void SetThreadCount(int count)
        {
            _thread_count = count;
        }

        void Create()
        {
            if(_thread_count>0)
            {
                _threads.resize(_thread_count);
                _loops.resize(_thread_count);
                for(int i = 0;i<_thread_count;i++)
                {
                    _threads[i] = new LoopThread();
                    _loops[i] = _threads[i]->GetLoop();
                }
            }
            return;
        }

        EventLoop* NextLoop()
        {
            if(_thread_count==0)
                return _baseloop;
            _next_idx = (_next_idx+1)%_thread_count;
            return _loops[_next_idx];
        }

};




class Any
{
    private:

        class holder
        {
            public:
                virtual ~holder() {}
                virtual const std::type_info& type() = 0;
                virtual holder* clone() = 0;
        };

        template<class T>
        class placeholeder :public holder
        {
            public:
                placeholeder(const T& val):
                _val(val){}

                virtual const std::type_info& type() //获取保存的数据类型
                {
                    return typeid(T);
                }
                virtual holder* clone()//针对当前的对象自身 克隆出一个新的子类对象
                {
                    return new placeholeder(_val);
                }

            public:
                T _val;
        };

        holder* _content;
    
    public:
        Any():_content(nullptr){}
        template<class T>
        Any(const T &val):_content(new placeholeder<T>(val)){}
        Any(const Any &other):_content(other._content ? other._content->clone() : nullptr){}

        ~Any(){delete _content;}

        Any &swap(Any& other)
        {
            std::swap(_content,other._content);
            return *this;
        }

        template<class T>
        T* get()//返回子类对象保存的数据的指针
        {
            assert(typeid(T)==_content->type());
            return  &((placeholeder<T>*)_content)->_val;
        }

        template<class T>
        Any& operator=(const T& val)
        {
            Any(val).swap(*this);
            return *this;
        }
        Any& operator=(const Any& other)
        {
            Any(other).swap(*this);
            return *this;
        }
};


class Connection;

//DISCONECTED连接关闭               CONNECTING连接建立成功待处理 
//CONNECTED连接建立完成 可以通信     DISCONNECTING待关闭状态
typedef enum{DISCONNECTED, CONNECTING, CONNECTED, DISCONNECTING}ConnStatu;

using PtrConnection = std::shared_ptr<Connection>;

class Connection : public std::enable_shared_from_this<Connection>
{
    private:
        uint64_t _conn_id;                  //连接的唯一id
        //uint64_t _timer_id                //定时器ID 必须唯一，使用_conn_id作为id来简化
        int _sockfd;                        //连接关联的文件描述符
        bool _enable_inactive_release;      //连接是否启动非活跃销毁的判断标志，默认false
        EventLoop *_loop;                   //连接所关联的eventloop
        ConnStatu _statu;                   //连接状态
        Socket _socket;                     //套接字操作管理
        Channel _channel;                   //连接的事件管理
        Buffer _in_buffer;                  //输入缓冲区----存放从socket中读取到的数据
        Buffer _out_buffer;                 //输出缓冲区----存放要发送给对端的数据
        Any _context;                       //请求的接受处理上下文

        //这四个回调函数，是让服务器模块来设置的 （其实是组件使用者设置给服务器模块的）
        //都是组件使用者设计的
        using ConnectedCallback = std::function<void(const PtrConnection&)>;
        using MessageCallback = std::function<void(const PtrConnection&, Buffer*)>;
        using ClosedCallback = std::function<void(const PtrConnection&)>;
        using AnyEventCallback = std::function<void(const PtrConnection&)>;

        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        ClosedCallback _closed_callback;
        AnyEventCallback _event_callback;

        //组件内的连接关闭回调 组件内设置的 因为服务器组件内会把所有连接管理，一旦某个连接要关闭
        //应该从管理的地方移除掉自己的信息
        ClosedCallback _server_closed_callback;

    private:
        //五个channel事件的回调函数/

        //描述符触发可读事件后调用的函数，接收socket数据放到接收缓冲区中，然后调用_message_callback
        void HandleRead()
        {
            //读取socket数据放到缓冲区
            char buf[65536];
            ssize_t ret = _socket.NonBlockRecv(buf,65535);
            if(ret <= 0)
            {
                //出错了不能直接关闭连接
                return ShutdownInLoop();
            }
            //0表示的是没有读取到数据，连接断开是-1
            //将数据放入输入缓冲区并将写偏移向后移动
            _in_buffer.WriteAndPush(buf,ret);
            //调用cb进行业务处理
            if(_in_buffer.ReadAbleSize()>0)
            {
                //shared_from_this  从当前对象自身获取自身的shared_ptr
                return _message_callback(shared_from_this(),&_in_buffer);
            }
        }

        //描述符触发可写事件后调用的函数，将发送缓冲区中的数据进行发送
        void HandleWrite()
        {
            //_out_buffer保存的数据就是要发送的数据
            ssize_t ret = _socket.NonBlockSend(_out_buffer.ReadPosition(),_out_buffer.ReadAbleSize());
            if(ret<0)
            {
                //发送错误 应该关闭连接
                if(_in_buffer.ReadAbleSize()>0)
                {
                    _message_callback(shared_from_this(),&_in_buffer);
                }
                //实际的关闭释放操作
                return Release();
            }
            //将读偏移向后移动
            _out_buffer.MoveReadOffset(ret);

            if(_out_buffer.ReadAbleSize()==0)
            {
                //没有数据待发送 关闭写事件监控
                _channel.DisableWrite();

                //如果当前是连接待关闭状态 若有数据 发送完数据释放连接 没有数据直接释放
                if(_statu==DISCONNECTING)
                {
                    return Release();
                }
            }

            return;
        }

        //描述符触发挂断事件
        void HandleClose()
        {
            //一旦连接挂断了 套接字什么都干不了 因此有数据发送一下 然后直接关闭
            if(_in_buffer.ReadAbleSize()>0)
            {
                _message_callback(shared_from_this(),&_in_buffer);
            }
            return Release();
        }

        //描述符触发错误事件
        void HandleError()
        {
            return HandleClose();
        }

        //描述符触发任意事件
        void HandleEvent()
        {
            //刷新连接活跃度 延迟销毁定时任务
            //调用组件使用者的任意事件回调
            if(_enable_inactive_release==true)
            {
                _loop->TimerRefresh(_conn_id);
            }

            if(_event_callback)
            {
                _event_callback(shared_from_this());
            }
        }

        //连接获取之后，所处的状态下要进行各种设置（启动读监控 调用回调函数）
        void EstablishedInLoop()
        {
            //修改连接状态 启动读事件监控 调用回调函数
            assert(_statu == CONNECTING);//当前状态必须是上层的半连接状态
            _statu = CONNECTED;//当前函数执行完毕 则连接进入已完成连接状态

            //一旦启动读事件监控就有可能立即触发读事件 如果这时候启动了非活跃连接销毁
            _channel.EnableRead();
            if(_connected_callback)
                _connected_callback(shared_from_this());
        }

        //实际的释放接口
        void ReleaseInLoop()
        {
            //修改连接状态 将其设置为DISCONNCTED
            _statu = DISCONNECTED;

            //移除连接的事件监控
            _channel.Remove();
    
            //关闭描述符
            _socket.Close();

            //如果当前定时器队列中还有定时销毁任务 则取消任务
            if(_loop->HasTimer(_conn_id))
                CancelInactiveReleaseInLoop();

            //调用关闭回调函数 避免先移除服务器端的会导致Connection被释放导致出错 先用户释放
            if(_closed_callback)
                _closed_callback(shared_from_this());
            //移除服务器内部管理的连接信息
            if(_server_closed_callback)
                _server_closed_callback(shared_from_this());

        }

        //这个接口并不是实际的发送接口 而只是把数据放到了发送缓冲区 启动了可写事件监控
        void SendInLoop(Buffer &buf)
        {
            if(_statu == DISCONNECTED)return;
                       
            _out_buffer.WriteBufferAndPush(buf);
            
            if(_channel.WriteAble()==false)
            {
                _channel.EnableWrite();
            }
        }

        //这个关闭操作并非实际的连接释放操作 需要判断还有没有数据待处理 待发送
        void ShutdownInLoop()
        {
            //设置连接为半关闭状态
            _statu = DISCONNECTING;

            if(_in_buffer.ReadAbleSize()>0)
            {
                if(_message_callback)
                    _message_callback(shared_from_this(),&_in_buffer);
            }

            //要么写入数据的时候出错关闭 要么就是没有待发送数据直接关闭
            if(_out_buffer.ReadAbleSize()>0)
            {
                if(_channel.WriteAble()==false)
                {
                    _channel.EnableWrite();
                }
            }
            
            if(_out_buffer.ReadAbleSize()==0)
            {
                Release();
            }
        }

        //启动非活跃连接释放规则
        void EnableInactiveReleaseInLoop(int sec)
        {
            //将判断标准 _enable_inactive_release 设置为true
            _enable_inactive_release = true;
            //如果当前定时销毁任务已存在则刷新 
            if(_loop->HasTimer(_conn_id))
                return _loop->TimerRefresh(_conn_id);
            //不存在则添加
            _loop->TimerAdd(_conn_id,sec,std::bind(&Connection::Release,this));
        }

        void CancelInactiveReleaseInLoop()
        {
            _enable_inactive_release = false;
            if(_loop->HasTimer(_conn_id))
                _loop->TimerCancel(_conn_id);
        }

        void UpgradeInLoop(const Any &context,
                           const ConnectedCallback &conn,
                           const MessageCallback &msg,
                           const ClosedCallback &closed,
                           const AnyEventCallback &event)
        {
            _context = context;
            _connected_callback = conn;
            _message_callback = msg;
            _closed_callback = closed;
            _event_callback = event;
        }

    public:
        Connection(EventLoop *loop,uint64_t conn_id,int sockfd):_conn_id(conn_id),_sockfd(sockfd),
        _enable_inactive_release(false),_loop(loop),_statu(CONNECTING),_socket(_sockfd),
        _channel(_loop,_sockfd)
        {
            _channel.SetCloseCallback(std::bind(&Connection::HandleClose,this));
            _channel.SetEventCallback(std::bind(&Connection::HandleEvent,this));
            _channel.SetReadCallback(std::bind(&Connection::HandleRead,this));
            _channel.SetWriteCallback(std::bind(&Connection::HandleWrite,this));
            _channel.SetErrorCallback(std::bind(&Connection::HandleError,this));
        }

        ~Connection()
        {
            DBG_LOG("RELEASE CONNECTION:%p",this);
        }

        //获取管理的文件描述符
        int Fd()
        {
            return _sockfd;
        }

        //获取连接的id
        int Id()
        {
            return _conn_id;
        }

        //返回状态 是否处于Connected状态
        bool Connected()
        {
            return _statu==CONNECTED;
        }

        //设置上下文 连接建立完成时
        void SetContext(const Any& context)
        {
            _context = context;
        }

        //获取上下文    返回指针
        Any *GetContext()
        {
            return &_context;
        }

        void SetConnectedCallback(const ConnectedCallback &cb)
        {
             _connected_callback = cb;
        }

        void SetMessageCallback(const MessageCallback &cb)
        {
            _message_callback = cb;
        }

        void SetClosedCallback(const ClosedCallback &cb)
        {
            _closed_callback = cb;
        }

        void SetAnyEventCallback(const ConnectedCallback &cb)
        {
            _event_callback = cb;
        }

        void SetSrvClosedCallback(const ClosedCallback &cb)
        {
            _server_closed_callback = cb;
        }

        //连接建立就绪后 进行channel回调设置，启动读监控，调用connected_callback
        void Established()
        {
            _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
        }

        //发送数据将数据放到发送缓冲区，启动写事件监控
        void Send(const char* data,size_t len)
        {
            //外界传入的data可能是临时空间 我们只把操作压入了任务池 没有被立即执行
            //执行时data可能被释放了
            Buffer buf;
            buf.WriteAndPush(data,len);
            _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,std::move(buf)));
        }

        //提供给最组件使用者的关闭接口--并不直接关闭 需要判断
        void Shutdown()
        {
            _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop,this));
        }

        void Release()
        {
            _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop,this));
        }

        
        //启动非活跃销毁，并定义多长时间无通信就是非活跃，添加定时任务
        void EnableInactiveRelease(int sec)
        {
             _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
        }

        //取消非活跃销毁
        void CancelInactiveRelease()
        {
            _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop,this));
        }

        //切换协议 重置上下文以及阶段性处理函数  需要在eventloop同一线程下 立即执行
        //防止新事件触发后使用元数据协议处理
        void Upgrade(const Any & context,const ConnectedCallback &conn,const MessageCallback &msg,
                     const ClosedCallback &closed,const AnyEventCallback &event)
        {
            _loop->AssertInLoop();
            _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,conn,msg,closed,event));
        }
};


class Acceptor
{
    private:
        Socket _socket;         //用于创建监听套接字
        EventLoop* _loop;       //用于对监听套接字进行事件监控
        Channel _channel;       //用于对监听套接字进行事件管理

        using AcceptCallback = std::function<void(int)>;
        AcceptCallback _accept_callback;

    private:
        //监听套接字的读事件回调函数 获取新连接 调用_accept_callback来连接新请求
        void HandleRead()
        {
            int newfd = _socket.Accept();
            if(newfd<0)
                return;

            if(_accept_callback)
                _accept_callback(newfd);
        }

        int CreateServer(int port)
        {
            bool ret = _socket.CreateServer(port);
                assert(ret==true);
            return _socket.Fd();
        }

    public:
        //不能将启动读事件监控放到构造函数中，否则有可能造成启动监控后立即有事件
        //可能新连接得不到处理 资源泄漏
        Acceptor(EventLoop* loop,int port):_socket(CreateServer(port)),_loop(loop),_channel(loop,_socket.Fd())
        {
            _channel.SetReadCallback(std::bind(&Acceptor::HandleRead,this));
        }

        void SetAcceptCallback(const AcceptCallback &cb)
        {
            _accept_callback = cb;
        }

        void Listen()
        {
            _channel.EnableRead();  
        }


};


class TcpServer
{
    private:
        uint64_t _next_id;              //自动增长的连接id
        int _port;                      //端口
        int _timeout;                   //非活跃连接的统计时间 多长时间无通信的时间
        bool _enable_inactive_release;  //是否启用非活跃销毁连接
        EventLoop _baseloop;            //这是主线程的EventLoop，负责监听事件的处理
        Acceptor _acceptor;             //监听套接字的管理对象
        LoopThreadPool _pool;           //从属Eventloop线程池
        std::unordered_map<uint64_t,PtrConnection> _conns;  //保存管理所有连接对应的shared_ptr对象

        using ConnectedCallback = std::function<void(const PtrConnection&)>;
        using MessageCallback = std::function<void(const PtrConnection&, Buffer*)>;
        using ClosedCallback = std::function<void(const PtrConnection&)>;
        using AnyEventCallback = std::function<void(const PtrConnection&)>;
        using Functor = std::function<void()>;

        ConnectedCallback _connected_callback;
        MessageCallback _message_callback;
        ClosedCallback _closed_callback;
        AnyEventCallback _event_callback;

    private:
        void RunAfterInLoop(const Functor& task,int delay)
        {
            _next_id++;
            _baseloop.TimerAdd(_next_id,delay,task);
        }

        //为新连接构造一个connection
        void NewConnection(int fd)
        {
            _next_id++;
            PtrConnection conn (new Connection(_pool.NextLoop(),_next_id,fd));
            conn->SetConnectedCallback(_connected_callback);
            conn->SetMessageCallback(_message_callback);
            conn->SetClosedCallback(_closed_callback);
            conn->SetAnyEventCallback(_event_callback);

            conn->SetSrvClosedCallback(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));

            if(_enable_inactive_release)
                conn->EnableInactiveRelease(_timeout);
            conn->Established();
            _conns.insert(std::make_pair(_next_id,conn));
        }

        void RemoveConnectionInLoop(const PtrConnection& conn)
        {
            int id = conn->Id();
            auto it  = _conns.find(id);
            if(it!=_conns.end())
                _conns.erase(it);
        }

        //从管理connection的_conns中移除连接信息
        void RemoveConnection(const PtrConnection& conn)
        {
            _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
        }

    public:
        TcpServer(int port):
        _port(port),
        _next_id(0),
        _enable_inactive_release(false),
        _acceptor(&_baseloop,port),
        _pool(&_baseloop)
        {
            _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
            //将监听套接字挂到baseloop上开始监控事件
            _acceptor.Listen();
        }


        void SetThreadCount(int count)
        {
            return _pool.SetThreadCount(count);
        }

        void SetConnectedCallback(const ConnectedCallback &cb)
        {
             _connected_callback = cb;
        }

        void SetMessageCallback(const MessageCallback &cb)
        {
            _message_callback = cb;
        }

        void SetClosedCallback(const ClosedCallback &cb)
        {
            _closed_callback = cb;
        }

        void SetAnyEventCallback(const ConnectedCallback &cb)
        {
            _event_callback = cb;
        }

        void EnableInactiveRelease(int timeout)
        {
            _timeout = timeout;
            _enable_inactive_release = true;

        }

        //用于添加一个定时任务
        void RunAfter(const Functor& task,int delay)
        {
            _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
        }

        void Start()
        {
            //创建从属线程
            _pool.Create();
            _baseloop.Start();
        }

};



void Channel::Remove()
{
    return _loop->RemoveEvent(this);
}   

void Channel::Update()
{
    return _loop->UpdateEvent(this);
}

void TimerWheel::TimerAdd(uint64_t id,uint32_t delay,const TaskFunc &cb)//考虑线程安全 _timers的操作有可能在多线程中进行，不想加锁就放在同一个线程中
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,id,delay,cb));
}

void TimerWheel::TimerRefresh(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));
}

void TimerWheel::TimerCancel(u_int64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,id));
}



class NetWork
{
    public:
        NetWork()
        {
            DBG_LOG("SIGPIPE INIT");
            signal(SIGPIPE,SIG_IGN);
        }
};

static NetWork nw;

#endif